/*
 * Copyright (c) 2017 Baidu, Inc. All Rights Reserve.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.baidu.fsg.uid.worker;

import com.baidu.fsg.uid.utils.DockerUtils;
import com.baidu.fsg.uid.utils.NetUtils;
import com.baidu.fsg.uid.worker.dao.WorkerNodeDAO;
import com.baidu.fsg.uid.worker.entity.WorkerNodeEntity;
import org.apache.commons.lang.math.RandomUtils;
import org.apache.commons.lang.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.Objects;
import java.util.concurrent.ThreadLocalRandom;

/**
 * Represents an implementation of {@link WorkerIdAssigner},
 * the worker id will be discarded after assigned to the UidGenerator
 *
 * @author yutianbao
 */
public class DisposableWorkerIdAssigner implements WorkerIdAssigner {
    private static final Logger LOGGER = LoggerFactory.getLogger(DisposableWorkerIdAssigner.class);

    @Autowired
    private WorkerNodeDAO workerNodeDAO;

    /**
     * Assign worker id base on database.<p>
     * If there is host name & port in the environment, we considered that the node runs in Docker container<br>
     * Otherwise, the node runs on an actual machine.
     *
     * @return assigned worker id
     */
    @Transactional(rollbackFor = Exception.class)
    public long assignWorkerId() {
        // build worker node entity
        WorkerNodeEntity workerNodeEntity = buildWorkerNode();

        // add worker node for new (ignore the same IP + PORT)
        workerNodeDAO.addWorkerNode(workerNodeEntity);
        LOGGER.info("Add worker node:" + workerNodeEntity);

        return workerNodeEntity.getId();
    }

    @Override
    public long assignReuseWorkerId(Long workerLevel1, Long workerLevel2, long maxWorkerId, Integer maxNoBeatDay) {
        Objects.requireNonNull(workerLevel1);
        Objects.requireNonNull(workerLevel2);
        Objects.requireNonNull(maxNoBeatDay);
        if (maxWorkerId <= 0) {
            throw new IllegalArgumentException();
        }


        /*
        怎样确定本次启动是新建WorkerID还是回收利用旧的WorkerID。
        简单。设置两个WorkID阈值,这里称作w1,w2, 当表中最大WorkID小于w1时一律新建WorkerID;
        当表中最大WorkID处于w1~w2之间时掷硬币方式确定是新建WorkerID还是回收利用旧的WorkerID;
        当表中最大WorkID大于w2时一律回收利用旧的WorkerID;
         */

        Long maxId = workerNodeDAO.selectMaxId();
        if (null == maxId) {
            maxId = 0L;
        }


        if (maxId < workerLevel1) {
            return assignWorkerId();
        }

        if (maxId >= workerLevel1 && maxId < workerLevel2) {

            if (coinNew()) {

                return assignWorkerId();
            } else {
                return reuseWorkerId(maxNoBeatDay);
            }

        } else {
            // maxId >= workerLevel2
            return assignWorkerId();
        }
    }

    /**
     * reuse worker node entity by IP and PORT
     */
    private long reuseWorkerId(Integer maxNoBeatDay) {

        Date now = new Date();
        Date minBeatDay = DateUtils.addDays(now, -maxNoBeatDay);

        while (true) {
            WorkerNodeEntity node = workerNodeDAO.selectMinIdForReuse(minBeatDay);
            if (null == node) {
                return assignWorkerId();
            }
            Integer reuseCount = node.getReuseCount();
            int effectRows = workerNodeDAO.reuseWorker(node.getId(), reuseCount, now);
            if (effectRows > 0) {
                return node.getId();
            }
        }

    }

    private boolean coinNew() {
        ThreadLocalRandom random = ThreadLocalRandom.current();
        return random.nextInt() % 2 == 0;
    }


    /**
     * Build worker node entity by IP and PORT
     */
    private WorkerNodeEntity buildWorkerNode() {
        WorkerNodeEntity workerNodeEntity = new WorkerNodeEntity();
        if (DockerUtils.isDocker()) {
            workerNodeEntity.setType(WorkerNodeType.CONTAINER.value());
            workerNodeEntity.setHostName(DockerUtils.getDockerHost());
            workerNodeEntity.setPort(DockerUtils.getDockerPort());

        } else {
            workerNodeEntity.setType(WorkerNodeType.ACTUAL.value());
            workerNodeEntity.setHostName(NetUtils.getLocalAddress());
            workerNodeEntity.setPort(System.currentTimeMillis() + "-" + RandomUtils.nextInt(100000));
        }

        Date now = new Date();
        workerNodeEntity.setModified(now);
        workerNodeEntity.setCreated(now);
        workerNodeEntity.setBeatTime(now);
        workerNodeEntity.setLaunchDate(now);
        workerNodeEntity.setReuseCount(0);

        return workerNodeEntity;
    }

}
